package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo25Student {

  def main(args: Array[String]): Unit = {
    /**
      *
      * 4、统计偏科最严重的前100名学生  [学号，姓名，班级，科目，分数
      * 偏科程度判断依据：通过方差判断偏科程度
      *
      *
      */

    val conf: SparkConf = new SparkConf()
      .setMaster("local")
      .setAppName("student")


    val sc = new SparkContext(conf)


    val scoreRDD: RDD[String] = sc.textFile("data/score.txt")
    val courceRDD: RDD[String] = sc.textFile("data/cource.txt")


    /**
      * 1、由于每一个科目分数的范围不一样，需要对数据做标准化
      */
    val scoreKVRDD: RDD[(String, (String, Double))] = scoreRDD.map(line => {
      val split: Array[String] = line.split(",")
      (split(1), (split(0), split(2).toDouble))
    })

    val counrceKVRDD: RDD[(String, Double)] = courceRDD.map(line => {
      val split: Array[String] = line.split(",")
      val couId: String = split(0)
      val sumSco: Double = split(2).toDouble
      (couId, sumSco)
    })

    val joinRDD: RDD[(String, ((String, Double), Double))] = scoreKVRDD.join(counrceKVRDD)

    val biaoScoreRDD: RDD[(String, Double)] = joinRDD.map {
      case (couId: String, ((id: String, score: Double), sumScore: Double)) =>
        //分数标准化
        val sco: Double = score / sumScore
        (id, sco)
    }


    /**
      * 计算方差
      *
      */
    val scoGroupRDD: RDD[(String, Iterable[Double])] = biaoScoreRDD.groupByKey()

    val stdRDD: RDD[(String, Double)] = scoGroupRDD.map {
      case (id: String, scos: Iterable[Double]) =>

        val list: List[Double] = scos.toList
        //计算平均值
        val avg: Double = list.sum / list.length

        //计算分子
        val zi: Double = list.map(i => (i - avg) * (i - avg)).sum

        val std: Double = zi / list.length

        (id, std)
    }

    //排序取方差最大的前100个学生

    val sortByRDD: RDD[(String, Double)] = stdRDD.sortBy(_._2, false)


    val top10: Array[(String, Double)] = sortByRDD.take(100)

    //取出学号
    val top100Id: Array[String] = top10.map(_._1)


    top100Id.foreach(println)

  }

}
